package com.imooc.bilibili.service.config;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.imooc.bilibili.domain.UserFollowing;
import com.imooc.bilibili.domain.UserMoment;
import com.imooc.bilibili.domain.constant.UserMomentsConstant;
import com.imooc.bilibili.service.UserFollowingService;
//import com.imooc.bilibili.service.websocket.WebSocketService;
import io.netty.util.internal.StringUtil;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.core.RedisTemplate;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

//@Configuration
public class RocketMQConfig {

    @Value("${rocketmq.name.server.address}")  //在application-test中进行配置
    private String nameServerAddr;

    @Autowired  //引入redis工具类
    private RedisTemplate<String, String> redisTemplate;

    @Autowired
    private UserFollowingService userFollowingService;

    //定义生产者
    @Bean("momentsProducer")
    public DefaultMQProducer momentsProducer() throws Exception{
        DefaultMQProducer producer = new DefaultMQProducer(UserMomentsConstant.GROUP_MOMENTS); //新建一个常量类
        producer.setNamesrvAddr(nameServerAddr);  //名称服务器地址
        producer.start();   //启动
        return producer;
    }
   //定义消费者
    @Bean("momentsConsumer")  //实现推送的方式
    public DefaultMQPushConsumer momentsConsumer() throws Exception{
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(UserMomentsConstant.GROUP_MOMENTS);
        consumer.setNamesrvAddr(nameServerAddr);
        consumer.subscribe(UserMomentsConstant.TOPIC_MOMENTS, "*"); //定义的主题和二级主题
        consumer.registerMessageListener(new MessageListenerConcurrently() {  //新建一个监听器，完成推送
            //实现这个方法，参数，消息和上下文
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context){
               //最简单的方法，直接遍历输出信息
//                for(MessageExt msg : msgs) {
//                    System.out.println(msg);
//                }
//                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS
                MessageExt msg = msgs.get(0);
                if(msg == null){
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
                String bodyStr = new String(msg.getBody());   //取出的是byte数组
                UserMoment userMoment = JSONObject.toJavaObject(JSONObject.parseObject(bodyStr), UserMoment.class);//jsonobjet类似于一个mapper
                Long userId = userMoment.getUserId();   //获取userID，为了获取订阅与关注的信息
                List<UserFollowing>fanList = userFollowingService.getUserFans(userId);  //获取粉丝的列表
                for(UserFollowing fan : fanList){   //遍历粉丝列表，给每一个发送消息
                    String key = "subscribed-" + fan.getUserId();
                    String subscribedListStr = redisTemplate.opsForValue().get(key);  //通过redis获取订阅信息
                    List<UserMoment> subscribedList;  //先生成一个订阅列表
                    if(StringUtil.isNullOrEmpty(subscribedListStr)){
                        subscribedList = new ArrayList<>();
                    }else{
                        subscribedList = JSONArray.parseArray(subscribedListStr, UserMoment.class);  //将字符串装换成需要的列表，和转换的实体类的类型
                    }
                    subscribedList.add(userMoment);
                    redisTemplate.opsForValue().set(key, JSONObject.toJSONString(subscribedList));  //格式转换
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        consumer.start();  //启动
        return consumer;   //返回
    }

    @Bean("danmusProducer")
    public DefaultMQProducer danmusProducer() throws Exception{
        // 实例化消息生产者Producer
        DefaultMQProducer producer = new DefaultMQProducer(UserMomentsConstant.GROUP_DANMUS);
        // 设置NameServer的地址
        producer.setNamesrvAddr(nameServerAddr);
        // 启动Producer实例
        producer.start();
        return producer;
    }

//    @Bean("danmusConsumer")
//    public DefaultMQPushConsumer danmusConsumer() throws Exception{
//        // 实例化消费者
//        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(UserMomentsConstant.GROUP_DANMUS);
//        // 设置NameServer的地址
//        consumer.setNamesrvAddr(nameServerAddr);
//        // 订阅一个或者多个Topic，以及Tag来过滤需要消费的消息
//        consumer.subscribe(UserMomentsConstant.TOPIC_DANMUS, "*");
//        // 注册回调实现类来处理从broker拉取回来的消息
//        consumer.registerMessageListener(new MessageListenerConcurrently() {
//            @Override
//            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
//                MessageExt msg = msgs.get(0);
//                byte[] msgByte = msg.getBody();
//                String bodyStr = new String(msgByte);
//                JSONObject jsonObject = JSONObject.parseObject(bodyStr);
//                String sessionId = jsonObject.getString("sessionId");
//                String message = jsonObject.getString("message");
//                WebSocketService webSocketService = WebSocketService.WEBSOCKET_MAP.get(sessionId);
//                if(webSocketService.getSession().isOpen()){
//                    try {
//                        webSocketService.sendMessage(message);
//                    } catch (IOException e) {
//                        e.printStackTrace();
//                    }
//                }
//                // 标记该消息已经被成功消费
//                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
//            }
//        });
//        // 启动消费者实例
//        consumer.start();
//        return consumer;
//    }
}
